package com.myflink.day02;

import com.myflink.bean.WaterSensor;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.source.SourceFunction;

import java.util.Random;

/**
 * @author Shelly An
 * @create 2020/9/16 15:30
 */
public class Source_MySource {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        //1. 从自定义数据源读取

        DataStreamSource<WaterSensor> waterSensorDS = env.addSource(new MySourceFunction());
        waterSensorDS.print();

        env.execute();
    }

    /**
     * 自定义数据源：
     * 1. 实现SourceFunction
     * 2. 指定输出的类型 WaterSensor
     * 3. 重写两个方法run，cancel
     */
    public static class MySourceFunction implements SourceFunction<WaterSensor> {
        private boolean flag = true;

        /**
         * 数据执行的逻辑
         * @param ctx
         * @throws Exception
         */
        @Override
        public void run(SourceContext<WaterSensor> ctx) throws Exception {
            Random random = new Random();
            //源源不断的产生数据
            while (flag) {
                ctx.collect(
                        new WaterSensor(
                                //左闭右开
                                "sensor_" + random.nextInt(3),
                                System.currentTimeMillis(),
                                random.nextInt(10) + 40
                        )
                );
                Thread.sleep(2000L);
            }
        }

        /**
         * 取消
         */
        @Override
        public void cancel() {
            this.flag = false;
        }
    }
}
